package com.sheepone.distributor;

import com.lmax.disruptor.EventTranslator;
import com.sheepone.distributor.event.DistributeEvent;
import com.sheepone.distributor.event.EventContext;
import com.sheepone.util.time.getter.TimeStampGetter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Miss.杨
 * @description
 * @since 2024/3/23 - 22:43
 */
public class DelayDistributor extends Distributor {
    private static final Logger logger = LoggerFactory.getLogger(new Object() {
    }.getClass().getEnclosingClass());

    private final AtomicInteger FIVE_HUNDREDS_TIME = new AtomicInteger(0);

    private final ConcurrentSkipListMap<Long, EventTranslator<DistributeEvent>> events = new ConcurrentSkipListMap<>();

    private Thread thread = new Thread(() -> {
        long sleepTime = Long.MAX_VALUE;
        while (!this.stopped.get()) {
            try {
                TimeUnit.NANOSECONDS.sleep(sleepTime);
            } catch (InterruptedException e) {
                // no op
            }
            while (!events.isEmpty()) {
                Map.Entry<Long, EventTranslator<DistributeEvent>> entry = events.firstEntry();
                long time = entry.getKey();
                long now = TimeStampGetter.nanos();
                if (time <= now) {
                    distribute(entry.getValue());
                    events.remove(entry.getKey());
                } else {
                    sleepTime = time - now;
                    break;
                }
            }
            if (events.isEmpty()) {
                sleepTime = Long.MAX_VALUE;
            }
        }
    });

    public DelayDistributor(String name, int size, List<DistributorLine> lines) {
        super(name, size, lines);
    }

    @Override
    public Distributor start() {
        Distributor distributor = super.start();
        thread.start();
        return distributor;
    }

    public void stop(){
        thread.interrupt();
        super.stop();
    }

    @Override
    public void distributeDelay(String targetStation, String targetTopic, EventContext context, long targetMillisStamp) {
        events.put(
                // 降低键重复概率
                targetMillisStamp * 1000_000 + FIVE_HUNDREDS_TIME.incrementAndGet() % 500,
                (e, s) -> {
                    e.setContext(context);
                    e.setTargetStation(targetStation);
                    e.setTargetTopic(targetTopic);
                }
        );
        thread.interrupt();
    }

}
